package kafkaTest;

import java.util.Arrays;
import java.util.Properties;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

public class MyConsumer {
	public static void main(String[] args) {

		// 属性配置
		Properties properties = new Properties();
		// 服务
		properties.put("bootstrap.servers", "192.168.12.151:9092,192.168.12.152:9092,192.168.12.154:9092");
		// 序列化
		properties.put("key.deserializer", "org.apache.kafka.common.serialization.IntegerDeserializer");
		properties.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
		properties.put("group.id","hdfs");//组
		properties.put("enable.auto.commit","true");//自动提交位置
		properties.put("auto.commit.interval.ms", "1000");// 自动提交时间间隔
		properties.put("session.timeout.ms", "30000");// 会话超时时间
		String topic = "test";
		//构建消费者
		KafkaConsumer<Integer, String> consumer = new KafkaConsumer<Integer, String>(properties);
		//消费主题
		consumer.subscribe(Arrays.asList(topic));//自动生成集合
		try {
			while (true) {
				ConsumerRecords<Integer, String> poll = consumer.poll(1000);
				for (ConsumerRecord<Integer, String> record : poll) {
					System.out.println(
							//模板字符串
							String.format("topic:%s,key:%d,value:%s", record.topic(), record.key(), record.value()));
				}
			}
		} catch (Exception e) {
			e.printStackTrace();
		} finally {
			consumer.close();
		}
	}
}
